package cn.itcast.es;

import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.bulk.CreateOperation;
import co.elastic.clients.elasticsearch.indices.*;
import co.elastic.clients.elasticsearch.indices.ExistsRequest;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.conn.ssl.NoopHostnameVerifier;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.ssl.SSLContextBuilder;
import org.apache.http.ssl.SSLContexts;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;

import javax.net.ssl.SSLContext;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.security.KeyStore;
import java.security.cert.Certificate;
import java.security.cert.CertificateFactory;
import java.util.ArrayList;
import java.util.List;

public class ESClient {

    private static ElasticsearchClient client;
    private static ElasticsearchAsyncClient asyncClient;
    private static ElasticsearchTransport transport;



    public static void main(String[] args) throws Exception {
        // 初始化连接
        initEsConnection();

        // 操作索引
//        operationIndexLambda();
//        operationIndex();
        operationBulk();
    }


    private static void operationBulk() throws Exception {

        User user = new User();
        user.setId(1001);
        user.setName("张三");
        user.setAge(18);

        CreateRequest<User> build = new CreateRequest.Builder<User>()
                .index("atguigu")
                .id("1001")
                .document(user)
                .build();


        // 创建文档
        CreateResponse createResponse = client.create(build);
        System.out.println(createResponse);

        // 批量创建文档
        List<BulkOperation> operations = new ArrayList<BulkOperation>();
        for ( int i= 1;i <= 5; i++ ) {
            CreateOperation<User> builder = new CreateOperation.Builder<User>()
            .index("atguigu")
            .id("200" + i)
            .document(new User(2000 + i, "zhangsan" + i, 1000 + i*1000)).build();

            BulkOperation bulk = new BulkOperation.Builder().create(builder).build();
            operations.add(bulk);
        }

        BulkRequest bulkRequest = new BulkRequest.Builder().operations(operations).build();
        BulkResponse bulkResponse = client.bulk(bulkRequest);
        System.out.println("数据操作成功：" + bulkResponse);
         // 删除文档
        DeleteRequest deleteRequest = new DeleteRequest.Builder().index("atguigu").id("2001").build();
        client.delete(deleteRequest);

        transport.close();
    }


    private static void operationIndexLambda() throws IOException {

        // 获取索引客户端
        final ElasticsearchIndicesClient indices = client.indices();

        // 判断索引是否存在
        boolean flg = indices.exists(req -> req.index("atguigu")).value();
        if (flg) {
            System.out.println("索引存在");
        } else {
            // 创建索引
            CreateIndexResponse createIndexResponse = indices.create(req -> req.index("atguigu"));
            System.out.println(createIndexResponse);
        }
        // 查询索引

        IndexState atguigu = indices.get(req -> req.index("atguigu")).get("atguigu");
        System.out.println("查询到的："+atguigu);


        // 删除索引

        boolean acknowledged  = indices.delete(req -> req.index("atguigu")).acknowledged();
        System.out.println("删除索引："+acknowledged);


        transport.close();
    }

    private static void operationIndex() throws IOException {

        // 获取索引客户端
        final ElasticsearchIndicesClient indices = client.indices();

        // 判断索引是否存在
        ExistsRequest existsRequest = new ExistsRequest.Builder().index("atguigu").build();
        BooleanResponse exists = indices.exists(existsRequest);
        if (exists.value()) {
            System.out.println("索引存在");
        } else {
            // 创建索引
            CreateIndexRequest atguigu = new CreateIndexRequest.Builder().index("atguigu").build();
            CreateIndexResponse createIndexResponse = indices.create(atguigu);
            System.out.println(createIndexResponse);
        }
        // 查询索引
        GetIndexRequest getIndexRequest = new GetIndexRequest.Builder().index("atguigu").build();
        GetIndexResponse getIndexResponse = indices.get(getIndexRequest);
        IndexState atguigu = getIndexResponse.get("atguigu");
        System.out.println("查询到的："+atguigu);


        // 删除索引
        DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest.Builder().index("atguigu").build();
        DeleteIndexResponse deleteIndexResponse = indices.delete(deleteIndexRequest);
        boolean acknowledged = deleteIndexResponse.acknowledged();
        System.out.println("删除索引："+acknowledged);


        transport.close();
    }



    private static void initEsConnection() throws Exception{  //  获取客户端对象
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials("elastic", "wcSda75FWD+hYjhi8J_v"));
        Path caCertificatePath = Paths.get("E:\\Study\\ideaProject\\spring-cloud\\spring-es-server-8\\certs\\es-api-ca.crt");
        CertificateFactory factory =
                CertificateFactory.getInstance("X.509");
        Certificate trustedCa;
        try (InputStream is = Files.newInputStream(caCertificatePath)) {
            trustedCa = factory.generateCertificate(is);
        }
        KeyStore trustStore = KeyStore.getInstance("pkcs12");
        trustStore.load(null, null);
        trustStore.setCertificateEntry("ca", trustedCa);
        SSLContextBuilder sslContextBuilder = SSLContexts.custom()
                .loadTrustMaterial(trustStore, null);
        final SSLContext sslContext = sslContextBuilder.build();
        RestClientBuilder builder = RestClient.builder(
                new HttpHost("192.168.20.129", 9200, "https")).setHttpClientConfigCallback(new
                   RestClientBuilder.HttpClientConfigCallback() {
                       @Override
                       public HttpAsyncClientBuilder customizeHttpClient(
                               HttpAsyncClientBuilder httpClientBuilder) {
                           return httpClientBuilder.setSSLContext(sslContext)
                                   .setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE)
                                   .setDefaultCredentialsProvider(credentialsProvider);
                       }
                   });
        RestClient restClient = builder.build();
        transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
        // 同步客户端对象
        client = new ElasticsearchClient(transport);
        // 异步客户端对象
        asyncClient = new ElasticsearchAsyncClient(transport);

    }
}
